#ifndef __MASTER_REQUEST__
#define __MASTER_REQUEST__

#include <limits.h>
#include <stdio.h>
#include <stdlib.h>
#include <inttypes.h>
#include <string.h>
#include <errno.h>

#include "ursax.h"
#include "log.h"
#include "anysocket.h"
#include "protocol.h"
#include "networking.h"
#include "st.h"
#include "endpoints.h"

#ifdef __cplusplus
extern "C" {
#endif

extern uint32_t manager_ip;
extern uint16_t manager_port;

#define MK_CONNECTION(stfd)											\
	st_netfd_t stfd = get_a_manager_connection();					\
	if (stfd == NULL)												\
	{																\
		return MK_CONNECTION_ERROR(-1);								\
	}

int request_manager_create_chunk_location_(st_netfd_t stfd, struct MgrReqChunkLocate* req, struct CSEP* csids);
static inline int request_manager_create_chunk_location(uint32_t volumeid, uint64_t volume_size, uint64_t chunk_size, uint16_t n, struct CSEP* csids)
{
	struct MgrReqChunkLocate req;

	setup_header(&req.header, MGR_REQ_CHUNK_CREATE);
	req.volumeid = volumeid;
	req.volume_size = volume_size;
	req.chunk_size = size64_convert_to32(chunk_size);
	req.cs_cnt = n;
	req.chunk_index = UINT_MAX; // represents not specified
	MK_CONNECTION(stfd);
	int ret = request_manager_create_chunk_location_(stfd, &req, csids);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_chunkserver_up_(st_netfd_t stfd, struct MgrReqServerUp* req);
static inline int request_manager_chunkserver_up(uint32_t my_ip, uint16_t my_port)
{
	struct MgrReqServerUp req;

	setup_header(&req.header, MGR_REQ_SERVER_UP);
	req.csid.ip = my_ip;
	req.csid.port = my_port;
	req.upcode = ERROR_NOP_DEBUG;
	MK_CONNECTION(stfd);
	int ret = request_manager_chunkserver_up_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_move_chunk_(st_netfd_t stfd, struct MgrReqChunkMove* req);
static inline int request_manager_move_chunk(uint32_t volumeid, uint32_t index, uint32_t from_ip, uint16_t port)
{
	struct MgrReqChunkMove req;
	memset(&req, 0, sizeof(req));
	setup_header(&req.header, MGR_REQ_CHUNK_MOVE);
	req.byid = true;
	req.volumeid = volumeid;
	req.index = index;
	req.from_ip = from_ip;
	req.from_port = port;
	MK_CONNECTION(stfd);
	int ret = request_manager_move_chunk_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_report_chunk_error_(st_netfd_t stfd, struct MgrReqChunkError* req);
static inline int request_manager_report_chunk_error(uint32_t chunk_server_ip, uint16_t chunk_server_port, uint16_t op, int error_code)
{
	struct MgrReqChunkError req;

	setup_header(&req.header, MGR_REQ_SERVER_ERROR);
	req.csid.ip = chunk_server_ip;
	req.csid.port = chunk_server_port;
	req.operation = op;
	req.connection_error_code = CONNECTION_ERROR(error_code);
	req.operation_error_code = OPERATION_ERROR(error_code);
	MK_CONNECTION(stfd);
	int ret = request_manager_report_chunk_error_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_report_chunk_error2_(st_netfd_t stfd, struct MgrReqChunkError2* req);
static inline int request_manager_report_chunk_error2(uint32_t chunk_server_ip, uint16_t chunk_server_port, uint32_t volumeid, uint32_t index, uint16_t op, int error_code)
{
	struct MgrReqChunkError2 req;

	setup_header(&req.header, MGR_REQ_SERVER_ERROR2);
	req.csid.ip = chunk_server_ip;
	req.csid.port = chunk_server_port;
	req.id.volumeid = volumeid;
	req.id.index = index;
	req.operation = op;
	req.connection_error_code = CONNECTION_ERROR(error_code);
	req.operation_error_code = OPERATION_ERROR(error_code);
	MK_CONNECTION(stfd);
	int ret = request_manager_report_chunk_error2_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

// this is for ec
int request_manager_report_chunk_error3_(st_netfd_t stfd, struct MgrReqChunkError3* req);
static inline int request_manager_report_chunk_error3(uint32_t chunk_server_ip, uint16_t chunk_server_port, uint32_t volumeid, uint32_t index, int k, int n, uint64_t chunk_size, uint32_t stripe_size_shift, uint16_t op, int error_code)
{
	struct MgrReqChunkError3 req;

	setup_header(&req.header, MGR_REQ_SERVER_ERROR3);
	req.csid.ip = chunk_server_ip;
	req.csid.port = chunk_server_port;
	req.id.volumeid = volumeid;
	req.id.index = index;
	req.operation = op;
	req.connection_error_code = CONNECTION_ERROR(error_code);
	req.operation_error_code = OPERATION_ERROR(error_code);
	req.k = k;
	req.n = n;
	req.chunk_size = size64_convert_to32(chunk_size);
	req.stripe_size_shift = stripe_size_shift;
	MK_CONNECTION(stfd);
	int ret = request_manager_report_chunk_error3_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_create_volume_(st_netfd_t stfd, struct MgrReqVolumeCreate* req, char* name, char* desc);
static inline int request_manager_create_volume(uint64_t totalsize, uint64_t chunksize, uint16_t repnum, char* name, char* desc)
{
	struct MgrReqVolumeCreate req;

	setup_header(&req.header, MGR_REQ_VOLUME_CREATE);
	req.name.length = name ? strlen(name) : 0;
	req.desc.length = desc ? strlen(desc) : 0;
	req.size = totalsize ? totalsize : 1024*1024*1024;
	req.data.chunk_size = size64_convert_to32(chunksize ? chunksize : 64*1024*1024);
	req.data.create_chunks = true;
	req.data.prealloc = true;
	req.data.repnum = repnum ? repnum : 3;
	MK_CONNECTION(stfd);
	int ret = request_manager_create_volume_(stfd, &req, name, desc);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_open_volume_(st_netfd_t stfd, struct MgrReqVolumeOpen* req, char* clientid, char** jvolume);
static inline int request_manager_open_volume(uint32_t volumeid, bool readonly, char* clientid, char** jvolume)
{
	struct MgrReqVolumeOpen req;

	setup_header(&req.header, MGR_REQ_VOLUME_OPENC);
	req.volumeid = volumeid;
	req.readonly = readonly;
	LOG_DEBUG("old open volume");
	MK_CONNECTION(stfd);
	int ret = request_manager_open_volume_(stfd, &req, clientid, jvolume);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_open_volume_name_(st_netfd_t stfd, struct MgrReqVolumeOpenName* req, char* name, char* clientid, char** jvolume);
static inline int request_manager_open_volume_name(char* name, bool readonly, char* clientid, char** jvolume)
{
	struct MgrReqVolumeOpenName req;

	setup_header(&req.header, MGR_REQ_VOLUME_OPEN);
	req.byid = false;
	req.readonly = readonly;
	MK_CONNECTION(stfd);
	int ret = request_manager_open_volume_name_(stfd, &req, name, clientid, jvolume);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_open_volumeX_(st_netfd_t stfd, struct MgrReqVolumeOpen2* req, char* clientid, char** jvolume);
static inline int request_manager_open_volumeX(uint32_t volumeid, bool readonly, char* clientid, char** jvolume, int cmd)
{
	struct MgrReqVolumeOpen2 req;

	int major_version, minor_version, patch_version;
	get_version_nbs(&major_version, &minor_version, &patch_version);

	setup_header(&req.header, cmd);
	req.volumeid = volumeid;
	req.readonly = readonly;
	req.major_version = major_version;
	req.minor_version = minor_version;
	req.patch_version = patch_version;
	MK_CONNECTION(stfd);
	int ret = request_manager_open_volumeX_(stfd, &req, clientid, jvolume);
	st_netfd_close(stfd);
	return ret;
}

static inline int request_manager_open_volume2(uint32_t volumeid, bool readonly, char* clientid, char** jvolume)
{
	return request_manager_open_volumeX(volumeid, readonly, clientid, jvolume, MGR_REQ_VOLUME_OPEN2);
}

static inline int request_manager_open_volume3(uint32_t volumeid, bool readonly, char* clientid, char** jvolume)
{
	return request_manager_open_volumeX(volumeid, readonly, clientid, jvolume, MGR_REQ_VOLUME_OPEN3);
}

int request_manager_close_volume_(st_netfd_t stfd, struct MgrReqVolumeClose* req, char* clientid);
static inline int request_manager_close_volume(uint32_t volumeid, char* clientid)
{
	struct MgrReqVolumeClose req;

	setup_header(&req.header, MGR_REQ_VOLUME_CLOSE2);
	req.volumeid = volumeid;
	req.force = 0;
	req.byid = 1;
	MK_CONNECTION(stfd);
	int ret = request_manager_close_volume_(stfd, &req, clientid);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_resize_volume_(st_netfd_t stfd, struct MgrReqVolumeResize* req);
static inline int request_manager_resize_volume(uint32_t volumeid, uint64_t new_size, bool preallocation)
{
	struct MgrReqVolumeResize req;

	setup_header(&req.header, MGR_REQ_VOLUME_RESIZE);
	req.volumeid = volumeid;
	req.new_size = new_size;
	req.preallocation = preallocation;
	MK_CONNECTION(stfd);
	int ret = request_manager_resize_volume_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_volume_setformat_(st_netfd_t stfd, struct MgrReqVolumeSetfmt *req);
static inline int request_manager_volume_setformat(uint32_t volumeid, uint16_t fmt, uint64_t virtual_size)
{
	LOG_INFO("volumeid %d fmt %d virtual_size %ld", volumeid, fmt, virtual_size);
	struct MgrReqVolumeSetfmt req;

	setup_header(&req.header, MGR_REQ_VOLUME_SETFMT);
	req.volumeid = volumeid;
	req.new_fmt = fmt;
	req.byid = true;
	req.virtual_size = virtual_size;
	MK_CONNECTION(stfd);
	int ret = request_manager_volume_setformat_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_client_fault_(st_netfd_t stfd, struct MgrReqClientFault *req, char* clientid);
static inline int request_manager_client_fault(char* clientid, uint32_t volumeid)
{
	struct MgrReqClientFault req;
	setup_header(&req.header, MGR_REQ_CLIENT_FAULT);
	req.volumeid = volumeid;
	MK_CONNECTION(stfd);
	int ret = request_manager_client_fault_(stfd, &req, clientid);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_snapshot_AddAndUpdate_(st_netfd_t stfd, struct MgrReqSnapshotAdd *req, char *json_str);
static inline int request_manager_snapshot_AddAndUpdate(char *json_str, bool add)
{
	struct MgrReqSnapshotAdd req;
	setup_header(&req.header, add? MGR_REQ_SNAPSHOT_ADD: MGR_REQ_SNAPSHOT_UPDATE);
	MK_CONNECTION(stfd);
	int ret = request_manager_snapshot_AddAndUpdate_(stfd, &req, json_str);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_snapshot_delete_(st_netfd_t stfd, struct MgrReqSnapshotDelete *req);
static inline int request_manager_snapshot_delete(uint32_t volume_id, uint32_t id)
{
	struct MgrReqSnapshotDelete req;
	setup_header(&req.header, MGR_REQ_SNAPSHOT_DELETE);
	req.volume_id = volume_id;
	req.id = id;
	MK_CONNECTION(stfd);
	int ret = request_manager_snapshot_delete_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_snapshot_get_(st_netfd_t stfd, struct MgrReqSnapshotGet *req, char **msg_out);
static inline int request_manager_snapshot_get(uint32_t volume_id, char **msg_out)
{
	struct MgrReqSnapshotGet req;
	setup_header(&req.header, MGR_REQ_SNAPSHOT_GET);
	req.volume_id = volume_id;
	MK_CONNECTION(stfd);
	int ret = request_manager_snapshot_get_(stfd, &req, msg_out);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_volume_set_active_(st_netfd_t stfd, struct MgrReqVolumeSetActive *req);
static inline int request_manager_volume_set_active(uint32_t volume_id)
{
	struct MgrReqVolumeSetActive req;
	setup_header(&req.header, MGR_REQ_VOLUME_SETACTIVE);
	req.volumeid = volume_id;
	req.byid = 1;
	MK_CONNECTION(stfd);
	int ret = request_manager_volume_set_active_(stfd, &req);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_backup_AddAndUpdate_(st_netfd_t stfd, struct MgrReqBackupAdd *req, char *json_str);
static inline int request_manager_backup_AddAndUpdate(char *json_str, bool add)
{
	struct MgrReqBackupAdd req;
	setup_header(&req.header, add ? MGR_REQ_VOLUME_BACKUP_ADD: MGR_REQ_VOLUME_BACKUP_UPDATE);
	MK_CONNECTION(stfd);
	int ret = request_manager_backup_AddAndUpdate_(stfd, &req, json_str);
	st_netfd_close(stfd);
	return ret;
}

static inline int request_manager_backup_Add(char *json_str)
{
	return request_manager_backup_AddAndUpdate(json_str, true);
}

static inline int request_manager_backup_Update(char *json_str)
{
	return request_manager_backup_AddAndUpdate(json_str, false);
}

int request_manager_backup_delete_(st_netfd_t stfd, struct MgrReqBackupDelete *req, char *name);
static inline int request_manager_backup_delete(uint32_t volume_id, char *name, bool force)
{
	struct MgrReqBackupDelete req;
	setup_header(&req.header, MGR_REQ_VOLUME_BACKUP_DELETE);
	req.volume_id = volume_id;
	req.force = force;
	MK_CONNECTION(stfd);
	int ret = request_manager_backup_delete_(stfd, &req, name);
	st_netfd_close(stfd);
	return ret;
}

int request_manager_backup_get_(st_netfd_t stfd, struct MgrReqBackupGet *req, char **msg_out);
static inline int request_manager_backup_get(uint32_t volume_id, char **msg_out)
{
	struct MgrReqBackupGet req;
	setup_header(&req.header, MGR_REQ_VOLUME_BACKUP_GET);
	req.volume_id = volume_id;
	MK_CONNECTION(stfd);
	int ret = request_manager_backup_get_(stfd, &req, msg_out);
	st_netfd_close(stfd);
	return ret;
}


#define MAX_WAIT_TIME (16)
#define RETRY_REQUEST_MANAGER_TIMES(cmd, times) 										\
{																						\
	int ret;																			\
	for(uint64_t i = 0, wait_time = 0; i < times; i++){										\
		ret = cmd;																		\
		if(ret == 0){																	\
			if(i != 0){																	\
				LOG_INFO("retryed %ld times and success :"#cmd, i);						\
			}																			\
			return ret;																	\
		}																				\
		if(CONNECTION_ERROR(ret)){														\
			LOG_ERROR("connection error return sleep %ds manager %s:%d", wait_time, str_ip(manager_ip), manager_port);	\
			st_sleep(++wait_time);														\
			if(wait_time > MAX_WAIT_TIME){ wait_time = MAX_WAIT_TIME; }					\
			continue;																	\
		}																				\
		if(OPERATION_ERROR(ret)){														\
			LOG_WARN("operation error, manager ret error code, manager %s:%d", str_ip(manager_ip), manager_port);	\
			break;																		\
		}																				\
		LOG_FATAL("unknown ret");														\
		break;																			\
	}																					\
	LOG_WARN("retryed and finally failed ret %d", ret);									\
	return ret;																			\
}

#define RETRY_REQUEST_MANAGER(cmd) RETRY_REQUEST_MANAGER_TIMES(cmd, -1UL)

#undef MK_CONNECTION

#ifdef __cplusplus
}
#endif

#endif
